package com.tgy.kafka.controller;

import com.alibaba.fastjson.JSONObject;
import com.tgy.kafka.config.KafkaConfig;
import com.tgy.kafka.constant.Myconstant;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/***
 * @ClassName: SendController
 * @Description:
 * @Auther: tanggy3
 * @Date: 14:12 2022/6/25
 * @version : V1.0
 */
@RestController
@RequestMapping("/kafka")
public class SendController {
    //引入producer
    @Resource
    KafkaConfig kafkaConfig;



    @GetMapping("/send")
    public String send() throws ExecutionException, InterruptedException {
        ProducerFactory<String, Object> stringObjectProducerFactory = kafkaConfig.tgyProducerFactory();
        Producer<String, Object> producer = stringObjectProducerFactory.createProducer();
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("message","你好");
        Future<RecordMetadata> send = producer.send(new ProducerRecord<>(Myconstant.TEST_TOPIC_NAME, jsonObject.toJSONString()));
        RecordMetadata recordMetadata = send.get();
        System.out.println(recordMetadata);
        return Myconstant.SUCCESS;
    }
}
